package org.example.utils;

import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;

/**
 * @Author: tang
 * @Description:
 * @Date 2025/8/4 16:26
 */
@Slf4j
@Component
public class RocketMqUtil {

    private RocketMQTemplate rocketMQTemplate;

    public RocketMqUtil(RocketMQTemplate rocketMQTemplate) {
        this.rocketMQTemplate = rocketMQTemplate;
        DefaultMQProducer producer = this.rocketMQTemplate.getProducer();
        producer.setRetryTimesWhenSendAsyncFailed(3);
        producer.setSendMsgTimeout(3000);
    }

    /**
     * 简单异步发送
     *
     * @param topicName
     * @param msg
     */
    public void sendAsyncMsg(String topicName, String msg) {
        sendAsyncMsg(topicName, msg, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                log.info("sendAsyncMsg 发送成功: {}", sendResult);
            }

            @Override
            public void onException(Throwable throwable) {
                log.error("sendAsyncMsg 发送失败: ", throwable);
            }
        });
    }

    /**
     * 异步发送 自定义回调
     *
     */
    public void sendAsyncMsg(String topicName, String msg, SendCallback sendCallback) {
        Message<?> message = new TenantIdMessagePostProcessor().postProcessMessage(MessageBuilder.withPayload(msg).build());
        this.rocketMQTemplate.asyncSend(topicName, message,sendCallback);
    }


    /**
     * 简单同步发送
     *
     * @param topicName
     * @param msg
     * @return
     */
    public SendResult sendSyncMsg(String topicName, String msg) {
        Message<String> message = MessageBuilder.withPayload(msg).build();
        new TenantIdMessagePostProcessor().postProcessMessage(message);
        return this.rocketMQTemplate.syncSend(topicName, message);
    }

    /**
     * 延迟发送 等级
     *
     */
    public SendResult sendSyncMsgDelayLevel(String topicName, String msg, int delayLevel) {
        Message<String> message = MessageBuilder.withPayload(msg).build();
        new TenantIdMessagePostProcessor().postProcessMessage(message);
        return this.rocketMQTemplate.syncSendDelayTimeSeconds(topicName, message, delayLevel);
    }

    /**
     * 延迟发送 时间,单位秒
     *
     */
    public SendResult sendSyncMsgDelayTime(String topicName, String msg, long delayTime) {
        Message<String> message = MessageBuilder.withPayload(msg).build();
        new TenantIdMessagePostProcessor().postProcessMessage(message);
        return this.rocketMQTemplate.syncSendDelayTimeSeconds(topicName, message, delayTime);
    }

    public RocketMQTemplate getRocketMQTemplate() {
        return rocketMQTemplate;
    }
}
